#pragma once

#include <iostream>
#include <chrono>
#include <ctime>
#include <string>
#include <mutex>
#include <map>
#include <unordered_map>
#include <unordered_set>
#include <functional>
#include <concurrent_queue/concurrent_queue.hpp>
#include "../common/define.hpp"
#include "socket.hpp"
namespace shine{
	namespace net{
		class peer;
        class acceptor;
        class connection;
		class connector;
		class udp_acceptor;
		class udp_connector;
		
		class proactor_engine_base;

		typedef std::unordered_map<std::string, connection*> map_connection_t;
		typedef map_connection_t * map_connection_p_t;
		typedef proactor_engine_base *proactor_base_t;
		typedef timer_manager* timer_manager_t;
		typedef void* data_t;

        struct task_data_block_t
        {
            size_t task_len = 0;
            const int8* data = nullptr;
            size_t data_len = 0;
            bool flush = false;
        } ;

        typedef std::function<void(connection *conn)> close_callback_t;
        typedef std::function<bool(connection *conn)> timeout_callback_t;
        typedef std::function<bool(const int8 *data, size_t len, connection *conn)> recv_callback_t;
        typedef std::function<bool(size_t len, connection *conn)> send_callback_t;
		typedef std::function<bool(bool, connection *conn)> connection_callback_t;
		typedef std::function<bool(bool, connection *conn)> accept_callback_t;
		typedef std::function<bool(bool, connector *conn)> connect_callback_t;

		class proactor_engine_base {
		public:
			virtual ~proactor_engine_base(){

			}

			//interface
			virtual acceptor *add_acceptor(const string &name, const string &addr, accept_callback_t cb, bool reuse_addr_port = true, bool reaccept = true, uint32 reaccept_delay = 3000) {
				return nullptr;
			}

			virtual udp_acceptor *add_udp_acceptor(const string &name, const string &addr, accept_callback_t cb, bool reuse_addr_port = true) {
				return nullptr;
			}

			connector * add_connector(const string &name, const string &conn_addr, connect_callback_t cb, const string bind_addr = "0.0.0.0:0", bool reconnect = true, uint32 reconnect_delay = 5000, bool reuse_addr_port = true) {
				return nullptr;
			}

			virtual udp_connector* add_udp_connector(const string &name, const string &conn_addr, connect_callback_t cb, const string bind_addr = "0.0.0.0:0", bool reconnect = true, uint32 reconnect_delay = 5000) {
				return nullptr;
			}

			virtual connection* add_connection(const string &name, socket_t fd, connection_callback_t cb) {
				return nullptr;
			}

			bool need_async() {
				if (empty_tid == tid)
				{
					return false;
				}

				return tid != std::this_thread::get_id();
			}

			void set_tid(const std::thread::id &other) {
				tid = other;
			}

			virtual void queue_execute(std::function<void()> &cmd) {
				if (need_async())
				{
					command_que.push(cmd);
					::send(command_fd.second, "", 1, 0);
				}
				else {
					cmd();
				}
			}

			void add_peer(peer *obj) {
				peers.insert(obj);
			}

			void del_peer(peer *obj) {
				peers.erase(obj);
			}

			bool check_peer(peer *obj) {
				return peers.find(obj) != peers.end();
			}
		protected:
			const std::thread::id empty_tid;
			std::thread::id tid;
			SHINE_GEN_MEMBER_GETSET(timer_manager, timer);
			SHINE_GEN_MEMBER_GETSET(bool, stop_flag, = false);

			typedef concurrent_queue<std::function<void()> >command_que_t;
			command_que_t command_que;
			std::pair<socket_t, socket_t> command_fd;
			std::unordered_set<peer*> peers;
		};

		enum status_t {
			e_idle = 1,
			e_accept = 2,
			e_connect = 4,
			e_recv = 8,
			e_send = 16,
			e_close = 32,
			e_exit = 64,
			e_udp_accept = 128,
			e_udp_connect = 256,
		};

		enum peer_type_t {
			e_acceptor = 0,
			e_connector = 1,
			e_connection = 2,
			e_udp_acceptor = 3,
			e_udp_connector = 4,
			e_udp_connection = 5
		};


        struct context
#if (defined SHINE_OS_WINDOWS)
            : OVERLAPPED
#endif
        {
            context()
            {
#if (defined SHINE_OS_WINDOWS)
                memset(this, 0, sizeof(OVERLAPPED));
                set_status(e_idle);
#endif

            }

            void flush(size_t len) {
                _buf.erase(0, len);
            }

            static const size_type _recv_some = 64*1024;
            const size_type get_recv_some() const{ return _recv_some; }

            SHINE_GEN_MEMBER_GETSET(shine::string, buf);

#if (defined SHINE_OS_WINDOWS)
            SHINE_GEN_MEMBER_GETSET(int32, status, = e_idle);
            SHINE_GEN_MEMBER_GETSET(WSABUF, WSABuf);
            
        public:
            typedef void* parent_t;
            SHINE_GEN_MEMBER_GETSET(parent_t, parent, = nullptr);
#endif

        };

        class proactor_engine;
        class peer {
        public:
            peer(){
#if (defined SHINE_OS_WINDOWS)
				get_recv_context().set_parent(this);
				get_send_context().set_parent(this);
#endif

				register_recv_callback([](const int8 *data, shine::size_t len, connection *conn)->bool {
					return true;
				});

				register_send_callback([](shine::size_t len, connection *conn)->bool {
					return true;
				});

				register_recv_timeout_callback([](connection *conn)->bool {
					return false;
				});

				register_send_timeout_callback([](connection *conn)->bool {
					return false;
				});

				register_close_callback([](connection *conn) {

				});

            }
            virtual ~peer(){
				if (get_engine())
				{
					get_engine()->del_peer(this);
				}
			}

/*
			bool default_recv_callback(const int8 *data, shine::size_t len, connection *conn) {
				return true;
			}

			bool default_send_callback(shine::size_t len, connection *conn){
				return true;
			}

			bool default_recv_timeout_callback(connection *conn) {
				return false;
			}

			bool default_send_timeout_callback(connection *conn) {
				return false;
			}

			void default_close_callback(connection *conn) {

			}
*/

			virtual void close() {};

		protected:
			virtual void shutdown() {
				socket::close(get_socket_fd());
				set_socket_fd(invalid_socket);

				stop_recv_timeout_timer();
				stop_send_timeout_timer();
				set_status(e_close);

			}

		public:
			virtual void async_recv() {}
			virtual void async_send(const int8 *data, size_t len, bool flush = true)
			{
				if (get_engine()->need_async())
				{
					std::string send_data(data, len);
					std::function<void()> func = [this, flush, send_data] {
						async_send(send_data.data(), send_data.size(), flush);
					};
					get_engine()->queue_execute(func);
					return;
				}

				if (!get_engine()->check_peer(this))
				{
					//log
					return;
				}


				iovec_t iov;
				iov.data = (int8 *)data;
				iov.size = len;

				async_sendv(&iov, 1, flush);
			}

			virtual void async_sendv(const iovec_t *iov, size_t count, bool flush = true) {}

			virtual bool async_connect() { return false; };

			virtual bool async_listen_accept() { return false; }

			void start_timer(uint32 timeout, uint64 &id, timeout_callback_t &cb) {
				stop_timer(id);

				if (timeout == 0)
					return;

				get_timer_manager()->cancel_timer(id);
				id = get_timer_manager()->set_timer(timeout, [this, &id, &cb]()->bool {
					if (!cb((connection*)this))
					{
						close();
						return false;
					}
					else
					{
						return true;
					}
				});
			}

			void stop_timer(uint64 &id)
			{
				get_timer_manager()->cancel_timer(id);
				id = invalid_timer_id;
			}
			void start_recv_timeout_timer() {
				start_timer(get_recv_timeout(), get_recv_timeout_timer_id(), get_recv_timeout_callback());
			}

			void stop_recv_timeout_timer() {
				stop_timer(get_recv_timeout_timer_id());
			}

			void start_send_timeout_timer() {
				start_timer(get_send_timeout(), get_send_timeout_timer_id(), get_send_timeout_callback());
			}

			void stop_send_timeout_timer() {
				stop_timer(get_send_timeout_timer_id());
			}

			void dump_error(const char* title) {
				std::cout << title << " fd:" << get_socket_fd() << " remote:" << get_remote_addr().get_address_string() << " local:" << get_local_addr().get_address_string() << " error:" << socket::get_error() << " " << socket::get_error_str(socket::get_error()) << std::endl;
			}

		protected:
			SHINE_GEN_MEMBER_GETSET(bool, v6, = false);
            SHINE_GEN_MEMBER_GETSET(string, name);
            SHINE_GEN_MEMBER_GETSET(socket_t, socket_fd, = invalid_socket);
			SHINE_GEN_MEMBER_GETSET(int32, type);
			SHINE_GEN_MEMBER_GETSET(bool, stop_flag, = false);
#if (defined SHINE_OS_WINDOWS)
            SHINE_GEN_MEMBER_GETSET(HANDLE, kernel_fd, = nullptr);
#else
            SHINE_GEN_MEMBER_GETSET(socket_t, kernel_fd, = invalid_socket);
#endif
        public:
			SHINE_GEN_MEMBER_GETSET(std::string, id);
			SHINE_GEN_MEMBER_GETSET(proactor_base_t, engine, = nullptr);
			SHINE_GEN_MEMBER_GETSET(timer_manager_t, timer_manager);
            SHINE_GEN_MEMBER_GETSET(data_t, bind_data, = nullptr);

			SHINE_GEN_MEMBER_GETSET(uint32, recv_timeout, = 0);
			SHINE_GEN_MEMBER_GETSET(uint32, send_timeout, = 0);
			SHINE_GEN_MEMBER_GETSET(uint64, recv_timeout_timer_id, = invalid_timer_id);
			SHINE_GEN_MEMBER_GETSET(uint64, send_timeout_timer_id, = invalid_timer_id);
			SHINE_GEN_MEMBER_GET_ONLY(context, recv_context);
			SHINE_GEN_MEMBER_GET_ONLY(context, send_context);
			SHINE_GEN_MEMBER_GETREG(accept_callback_t, accept_callback, = nullptr);
			SHINE_GEN_MEMBER_GETREG(connect_callback_t, connect_callback, = nullptr);
			SHINE_GEN_MEMBER_GETREG(close_callback_t, close_callback, = nullptr);
			SHINE_GEN_MEMBER_GETREG(timeout_callback_t, recv_timeout_callback, = nullptr);
			SHINE_GEN_MEMBER_GETREG(timeout_callback_t, send_timeout_callback, = nullptr);
			SHINE_GEN_MEMBER_GETREG(recv_callback_t, recv_callback, = nullptr);
			SHINE_GEN_MEMBER_GETREG(send_callback_t, send_callback, = nullptr);
			SHINE_GEN_MEMBER_GETSET(address_info_t, local_addr);
			SHINE_GEN_MEMBER_GETSET(address_info_t, remote_addr);
			SHINE_GEN_MEMBER_GETSET(int32, status, = e_idle);
			SHINE_GEN_MEMBER_GETSET(bool, reconnect, = false);
			SHINE_GEN_MEMBER_GETSET(uint32, reconnect_delay, = 5000);
			SHINE_GEN_MEMBER_GETSET(uint64, reconnect_timer_id, = invalid_timer_id);
			SHINE_GEN_MEMBER_GETSET(address_info_t, bind_addr);
			SHINE_GEN_MEMBER_GETSET(socket_t, new_conn_fd, = invalid_socket);
			SHINE_GEN_MEMBER_GETSET(string, bind_address);
			SHINE_GEN_MEMBER_GETSET(bool, reaccept, = false);
			SHINE_GEN_MEMBER_GETSET(uint64, reaccept_delay, = 3000);
			SHINE_GEN_MEMBER_GETSET(uint64, reaccept_timer_id, = invalid_timer_id);
			SHINE_GEN_MEMBER_GETSET(bool, reuse_addr_port, = true);
			SHINE_GEN_MEMBER_GETSET(map_connection_p_t, map_connection_p, = nullptr);
			SHINE_GEN_MEMBER_GETSET(bool, clear_map_connection, = false);
        };

    }
}
